#ifndef __PTHREADPOOL_H__
#define __PTHREADPOOL_H__

#include "common.h"
#include "singleton.h"
#include "type.h"
#include "log.h"
#include "epoll.h"
#include <pthread.h>
#include <queue>
#include <chrono>


typedef void(*HANDLE)(void*);

struct Task
{
	Task(int epollfd = 0, int sock = 0, void* event = nullptr)
	{
		epollfd_ = epollfd;
		fd_ = sock;
		event_ = event;
		state_ = 0;
		create_time_ = std::chrono::system_clock::now();
	}

	~Task()
	{
		fd_ = 0;
		epollfd_ = 0;
		event_ = nullptr;
		state_ = 0;
	}

  bool checkLiveTime(double t)
  {
    std::chrono::system_clock::time_point curTime = std::chrono::system_clock::now();
    std::chrono::duration<double> elapsed_seconds = curTime - create_time_;

    return elapsed_seconds.count() < t;
  }

  //int getTaskStatus()
  //{
	 // return state_;
  //}

  //void setTaskStatus(int st)
  //{
	 // pthread_mutex_lock(&mt_);
	 // state_ = st;
	 // pthread_mutex_unlock(&mt_);
  //}

  void run()
  {
    if(event_)
    {
      HANDLE func = (HANDLE)event_;
      func(this);
    }
  }
  int epollfd_;
  int fd_;
  void* event_;
  uint8 state_;
  std::chrono::system_clock::time_point create_time_;
};

class ThreadPool : public qSingleton<ThreadPool>
{
	friend class qSingleton<ThreadPool>;

private:
	ThreadPool()
	{

	}
	~ThreadPool()
	{
		pthread_mutex_destroy(&mt_);
		pthread_mutex_destroy(&num_mt_);
		pthread_cond_destroy(&cond_);
	}
	const ThreadPool& operator=(const ThreadPool&);
	ThreadPool(const ThreadPool&);

	void threadLock()
	{
		pthread_mutex_lock(&mt_);
	}
	void threadUnlock()
	{
		pthread_mutex_unlock(&mt_);
	}

	void threadWait()
	{
		pthread_cond_wait(&cond_, &mt_);
	}

	void threadSignal()
	{
		pthread_cond_signal(&cond_);
	}

	uint32 taskNum()
	{
		uint32 num = 0;
		std::map<int, std::queue<Task*> >::const_iterator cit = task_que_.begin();
		for (; cit != task_que_.end(); ++cit)
		{
			num += cit->second.size();
		}
		return num;
	}


	static void* run(void* arg)
	{
		Logger::info("run threadid:%u", pthread_self());
		ThreadPool* this_p = (ThreadPool*)arg;
		while (true)
		{
			Task* tk = this_p->getTask();
			if (!tk)
			{
				Logger::info("FFL[%s, %s, %u] tk is null threadid:%u, quesize:%u", LOG_FFL, pthread_self(), this_p->taskNum());
				this_p->threadUnlock();
				continue;
			}
			Logger::info("FFL[%s, %s, %u] get task tk:%p, epollid:%u, fd:%u, state:%u, threadid:%u, quesize:%u", LOG_FFL, tk, tk->epollfd_, tk->fd_, tk->state_, pthread_self(), this_p->taskNum());
			tk->run();

			EpollOpt::DelEpoll(tk->epollfd_, tk->fd_);
			close(tk->fd_);
			delete tk;
			tk = nullptr;
		}
		return nullptr;
	}
public:
	bool init(uint32 threadnum)
	{
		if (threadnum > max_thread_num_)
			threadnum = max_thread_num_;

		if (0 != pthread_mutex_init(&mt_, nullptr))
		{
			LOG_ERROR("pthread init mutex init fail");
			return false;
		}
		if (0 != pthread_mutex_init(&num_mt_, nullptr))
		{
			LOG_ERROR("pthread init num_mt_ init fail");
			return false;
		}

		if (0 != pthread_cond_init(&cond_, nullptr))
		{
			LOG_ERROR("pthread init mutex init fail");
			return false;
		}

		for (uint32 i = 0; i < threadnum; ++i)
		{
			pthread_t tid;
			if (0 != pthread_create(&tid, nullptr, run, this))
			{
				LOG_ERROR("pthread init create thread fail");
				return false;
			}
			pthread_detach(tid);
		}
		return true;
	}

	void setMaxThreadNum(uint32 num)
	{
		max_thread_num_ = num;
	}

	int getSockTaskNum(int sock)
	{
		int num = 0;
		threadLock();
		std::map<int, std::queue<Task*> >::const_iterator cit = task_que_.find(sock);
		if (cit != task_que_.end())
		{
			num = cit->second.size();
		}
		threadUnlock();
		return num;
	}

	void putTask(Task* tk)
	{
		threadLock();
		task_que_[tk->fd_].push(tk);
		Logger::info("FFL[%s, %s, %u] put task tk:%p, epollid:%u, fd:%u, state:%u threadid:%u, quesize:%u", LOG_FFL, tk, tk->epollfd_, tk->fd_, tk->state_, pthread_self(), task_que_.size());
		tk->state_ = 1;
		threadUnlock();
		//提醒有任务来到
		threadSignal();
	}

	Task* gettask()
	{
		Task* tk = nullptr;
		std::map<int, std::queue<Task*> >::iterator it = task_que_.begin();
		for (; it != task_que_.end(); ++it)
		{
			if (it->second.size() > 0)
			{

				tk = it->second.front();
				it->second.pop();
				tk->state_ = 2;
				if (it->second.empty())
				{
					task_que_.erase(it);
				}
				break;
			}
			else
			{
				task_que_.erase(it);
			}
		}
		return tk;
	}

	Task* getTask()
	{
		threadLock();
		while (taskNum() <= 0)
		{
			//LOG_INFO("cond thread %u", pthread_self());
			//没有任务，在此等待
			threadWait();
		}
		Task* tk = gettask();
		Logger::info("FFL[%s, %s, %u] get task epollid:%u, fd:%u, threadid:%u, quesize:%u", LOG_FFL, tk->epollfd_, tk->fd_, pthread_self(), taskNum());
		threadUnlock();
		return tk;
	}


	void addSockTaskNum(int sock)
	{
		pthread_mutex_lock(&num_mt_);
		sock_task_num_[sock]++;
		Logger::info("addSockTaskNum sock:%d, tasknum:%d", sock, sock_task_num_[sock]);
		pthread_mutex_unlock(&num_mt_);
	}

	void subSockTaskNum(int sock)
	{
		pthread_mutex_lock(&num_mt_);
		std::map<int, int>::iterator it = sock_task_num_.find(sock);
		if (it != sock_task_num_.cend())
		{
			if (it->second > 0)
			{
				it->second--;
				Logger::info("subSockTaskNum sock:%d, tasknum:%d", sock, sock_task_num_[sock]);
			}
			if (it->second == 0)
			{
				sock_task_num_.erase(it);
			}
		}
		pthread_mutex_unlock(&num_mt_);
	}

	int getSockDoingTaskNum(int sock)
	{
		int num = 0;
		pthread_mutex_lock(&num_mt_);
		std::map<int, int>::iterator it = sock_task_num_.find(sock);
		if (it != sock_task_num_.end())
		{
			num = it->second;
		}
		pthread_mutex_unlock(&num_mt_);
		Logger::info("getSockDoingTaskNum sock:%d, tasknum:%d", sock, num);
		return num;
	}
private:
	uint32 max_thread_num_;

	//std::queue<Task*> task_que_; 
	std::map<int, std::queue<Task*> > task_que_;

	//互斥量
	pthread_mutex_t mt_;
	//条件变量
	pthread_cond_t cond_;

	std::map<int, int> sock_task_num_;//sock包括正在执行task数量
	pthread_mutex_t num_mt_;

};


#endif
